package com.gitee.melin.bee.core.jdbc.dialect.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.ElasticsearchVersionInfo;
import co.elastic.clients.elasticsearch._types.mapping.TypeMapping;
import co.elastic.clients.elasticsearch.core.InfoResponse;
import co.elastic.clients.elasticsearch.indices.*;
import co.elastic.clients.elasticsearch.indices.get_mapping.IndexMappingRecord;
import co.elastic.clients.elasticsearch.indices.stats.IndexStats;
import co.elastic.clients.elasticsearch.indices.stats.IndicesStats;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import com.gitee.melin.bee.core.jdbc.QueryResult;
import com.gitee.melin.bee.core.jdbc.dialect.JdbcDialect;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.relational.*;
import com.gitee.melin.bee.util.DateUtils;
import java.sql.Statement;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import org.apache.commons.io.IOUtils;
import org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;

public class ElasticSearchDialect implements JdbcDialect {

    protected ConnectionInfo connectionInfo;

    private RestClient restClient;

    private ElasticsearchTransport transport;

    private ElasticsearchClient client;

    public ElasticSearchDialect(ConnectionInfo connectionInfo) {
        if (connectionInfo == null) {
            return;
        }

        this.connectionInfo = connectionInfo;

        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(
                AuthScope.ANY,
                new UsernamePasswordCredentials(connectionInfo.getUsername(), connectionInfo.getPassword()));

        restClient = RestClient.builder(HttpHost.create(connectionInfo.getEndpoint()))
                .setHttpClientConfigCallback(
                        httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider))
                .build();

        transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
        client = new ElasticsearchClient(transport);
    }

    @Override
    public DataSourceType getDataSourceType() {
        return DataSourceType.ELASTICSEARCH;
    }

    @Override
    public MetaDataSource testConnection(ConnectionInfo connectionInfo) {
        try (RestClient restClient = RestClient.builder(HttpHost.create(connectionInfo.getEndpoint()))
                        .build();
                ElasticsearchTransport transport = new RestClientTransport(restClient, new JacksonJsonpMapper())) {

            ElasticsearchClient client = new ElasticsearchClient(transport);
            InfoResponse infoResponse = client.info();
            ElasticsearchVersionInfo info = infoResponse.version();
            MetaDataSource metaDataSource = new MetaDataSource();
            metaDataSource.setDatabaseProductName(infoResponse.name());
            metaDataSource.setDatabaseProductVersion(info.number());
            metaDataSource.setBuildDate(
                    DateUtils.formatDateTime(info.buildDate().toInstant()));
            return metaDataSource;
        } catch (Exception e) {
            throw new RuntimeException("test es connection failed: " + e.getMessage(), e);
        }
    }

    @Override
    public DatabaseVersion getDatabaseVersion() {
        throw new UnsupportedOperationException();
    }

    @Override
    public List<String> getDatabases() {
        try {
            return Lists.newArrayList(client.info().clusterName());
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @Override
    public List<String> getSchemas() {
        return this.getDatabases();
    }

    @Override
    public List<String> getTableNames(String schemaName) {
        return getSchemaTables(schemaName).stream().map(MetaTable::getTableName).collect(Collectors.toList());
    }

    @Override
    public List<SimpleMetaTable> getSimpleTables(String schemaName) {
        return getSchemaTables(schemaName).stream()
                .map(metaTable -> {
                    SimpleMetaTable table = new SimpleMetaTable();
                    table.setTableName(metaTable.getTableName());
                    return table;
                })
                .collect(Collectors.toList());
    }

    @Override
    public MetaTable getSchemaTable(String schemaName, String tableName, boolean containColumn) {
        try {
            IndicesStatsResponse response = client.indices().stats();
            if (!response.indices().containsKey(tableName)) {
                throw new RuntimeException(tableName + " index not exists");
            }

            IndicesStats indicesStats = response.indices().get(tableName);
            IndexStats indexStats = indicesStats.total();

            MetaTable metaTable = new MetaTable();
            metaTable.setTableName(tableName);
            metaTable.setTableType(BeeTableType.TABLE);
            metaTable.setNumRows(indexStats.docs().count());
            metaTable.setDataSize(indexStats.store().totalDataSetSizeInBytes());

            if (containColumn) {
                List<MetaColumn> columns = this.getSchemaColumns(schemaName, tableName);
                metaTable.setColumns(columns);
            }
            return metaTable;
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @Override
    public List<MetaTable> getSchemaTables(String schemaName) {
        try {
            IndicesStatsResponse response = client.indices().stats();
            List<MetaTable> tables = Lists.newArrayList();
            response.indices().forEach((index, indicesStats) -> {
                MetaTable metaTable = new MetaTable();
                metaTable.setTableType(BeeTableType.TABLE);
                metaTable.setTableName(index);

                IndexStats indexStats = indicesStats.total();
                metaTable.setNumRows(indexStats.docs().count());
                metaTable.setDataSize(indexStats.store().totalDataSetSizeInBytes());
                tables.add(metaTable);
            });
            return tables;
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @Override
    public LinkedHashSet<String> getPrimaryKeys(String schemaName, String tableName) {
        throw new UnsupportedOperationException();
    }

    @Override
    public List<MetaColumn> getSchemaColumns(String schemaName, String tableName) {
        try {
            GetMappingRequest request = GetMappingRequest.of(re -> re.index(tableName));
            GetMappingResponse response = client.indices().getMapping(request);
            Map<String, IndexMappingRecord> indexMap = response.result();
            IndexMappingRecord record = indexMap.get(tableName);
            TypeMapping typeMapping = record.mappings();

            List<MetaColumn> columns = Lists.newArrayList();
            typeMapping.properties().forEach((key, property) -> {
                MetaColumn column = new MetaColumn();
                column.setColumnName(key);
                column.setDataType(property._kind().jsonValue());
                columns.add(column);
            });

            return columns;
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @Override
    public String getCreateTableScript(BeeTableType type, String schemaName, String tableName) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Statement getQueryStatement() {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String schema, String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String sql, int maxRecords) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String schema, String sql, int maxRecords) {
        throw new UnsupportedOperationException();
    }

    @Override
    public QueryResult query(String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public QueryResult query(String schema, String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public QueryResult query(String sql, int maxRecords) {
        throw new UnsupportedOperationException();
    }

    @Override
    public QueryResult query(String schema, String sql, int maxRecords) {
        throw new UnsupportedOperationException();
    }

    @Override
    public long queryCount(String sql, Object... params) {
        throw new UnsupportedOperationException();
    }

    @Override
    public List<Map<String, Object>> queryForList(String sql, Object... params) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Boolean execute(String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Boolean execute(String schema, String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<Boolean> asyncExecute(String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<Boolean> asyncExecute(String schema, String sql) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Integer executeUpdate(String sql, Object... params) {
        throw new UnsupportedOperationException();
    }

    @Override
    public Integer executeUpdate(SchemaIdentifier schema, String sql, Object... params) {
        return null;
    }

    @Override
    public CompletableFuture<Integer> asyncExecuteUpdate(String sql, Object... params) {
        throw new UnsupportedOperationException();
    }

    @Override
    public CompletableFuture<Integer> asyncExecuteUpdate(SchemaIdentifier schema, String sql, Object... params) {
        throw new UnsupportedOperationException();
    }

    @Override
    public void close() throws Exception {
        IOUtils.close(transport);
        IOUtils.close(restClient);
    }
}
